package com.seaboxdata.interfaceJob;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.robin.core.convert.util.ConvertUtil;
import com.seaboxdata.entity.GpsMessage;
import com.seaboxdata.kafka.BaseStringKafkaProducer;
import com.seaboxdata.utils.HttpUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.TimeUnit;

@Component
@Configuration
@Slf4j
public class GpsMessageCollectJob {

    private String accessToken = null;

    private Gson gson = new Gson();

    @Autowired
    private BaseStringKafkaProducer baseStringKafkaProducer;

    private DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private Cache<String, String> cache = CacheBuilder.newBuilder().initialCapacity(5000).expireAfterAccess(30, TimeUnit.MINUTES).build();

    @Autowired
    private AlarmMsgSynchronizer alarmMsgSynchronizer;

    @Value("${whp.gps.messageTopic}")
    private String topic;

    @Value("${connector.daxin.vehicle.gps.gpsGrepUrl}")
    private String gpsGrepUrl;

    @Value("${interface.isTest}")
    private String isTest;

    @Scheduled(cron = "${crontab.gpsmessage}")
    private void doGpsCollect() {
        //如果是测试,直接退出
        if("true".equals(isTest)){
            return;
        }
        log.info("collect gps message begin");
        //获取token
        if (null == accessToken) {
            accessToken = alarmMsgSynchronizer.doLoginRetToken();
        }
        Map<String, String> headerMap = new HashMap<>();
        headerMap.put("content-type", "application/json");
        headerMap.put("Authorization", accessToken);
        HttpUtils.Response response = HttpUtils.doGet(gpsGrepUrl, "UTF-8", headerMap);
        //token可能失效，重新获取
        if (response.getStatusCode() == 401) {
            accessToken = alarmMsgSynchronizer.doLoginRetToken();
            headerMap.put("Authorization", accessToken);
            response = HttpUtils.doGet(gpsGrepUrl, "UTF-8", headerMap);
        }

        if(response.getStatusCode()==200) {
            Map<String, Object> map = gson.fromJson(response.getResponseData(), new TypeToken<Map<String, Object>>() {
            }.getType());
            List<Map<String, Object>> list = (List<Map<String, Object>>) map.get("data");
            Date date = new Date();
            try {
                if (null != list && !list.isEmpty()) {
                    for (Map<String, Object> f : list) {
                        if (canInsert(f)) {
                            GpsMessage message = new GpsMessage();
                            ConvertUtil.mapToObject(message, f);
                            message.setCargo((String) f.get("huowu"));
                            message.setGenerateTime(new Timestamp(LocalDateTime.parse(f.get("positionTime").toString(), df).toEpochSecond(ZoneOffset.ofHours(8)) * 1000));
                            message.setCjsj(date);
                            //修改逻辑通过kafka,方便实时共享使用
                            baseStringKafkaProducer.sendMessageToKafka(JSONObject.toJSONString(message), topic);
                        }
                    }
                    log.info("危化品接口调用返回数据：{}", list.size());
                }else {
                    log.info("危化品接口调用返回数据：{}", JSON.toJSONString(response));
                }
            } catch (Exception ex) {
                log.error("{}", ex);
            }
        }else {
            log.info("危化品接口调用返回数据：{}", JSON.toJSONString(response));
        }
    }

    public boolean canInsert(Map<String, Object> map) {
        boolean canInsert = false;
        if (map.containsKey("carCode") && null != map.get("carCode")) {
            String tsStr = cache.getIfPresent(map.get("carCode").toString());
            if (null == tsStr) {
                canInsert = true;
                cache.put(map.get("carCode").toString(), map.get("positionTime").toString());
            } else if (!tsStr.equals(map.get("positionTime").toString())) {
                canInsert = true;
                cache.put(map.get("carCode").toString(), map.get("positionTime").toString());
            }
        }
        return canInsert;
    }
}
